Currently, AsyncBufferSequence will leak the base sequence if iteration concludes without next() being cancelled. This is because the internal task retains UnboundedBufferStorage/BoundedBufferStorage causing a reference cycle. This is fixed by adopting an InternalClass inside the iterator which calls an iteratorDeinitialized() function on the storage, as done by other sequences in this package such as combine latest and merge.

The following code illustrates the issue: the assert fails meaning the channel has leaked.

let channel = AsyncChannel<Void>()
            
Task {
    for await _ in channel.buffer(policy: .bounded(1)) {
        break
    }
    
    Task { [weak channel] in
        assert(channel == nil)
    }
}

Task { [weak channel] in
    await channel?.send(())
}